package com.xkstream.client.rx;

import io.reactivex.Flowable;
import io.reactivex.processors.FlowableProcessor;
import io.reactivex.processors.PublishProcessor;

/**
 * Created by xiaoyi on 2017/8/25.
 */

public class Rxbus {
    private static volatile Rxbus mDefault;
    private FlowableProcessor<Object> mBus;

    private Rxbus() {
        mBus = PublishProcessor.create().toSerialized();
    }

    public void post(Object o) {
        mBus.onNext(o);
    }

    public <T> Flowable<T> register(Class<T> clazz) {
        return mBus.ofType(clazz);
    }

    public void unRegisterAll() {
        mBus.onComplete();
    }

    public boolean hasSubscribers() {
        return mBus.hasSubscribers();
    }

    public static Rxbus getDefault() {
        if (mDefault == null) {
            synchronized (Rxbus.class) {
                if (mDefault == null) {
                    mDefault = new Rxbus();
                }
            }
        }
        return mDefault;
    }
}
